<?php
/* 
 * Functions for common reading and writing to Apache Cassandra database.
 *
 * Whenever an investment record is read from the database is must be adorned
 * with the following attributes: record_id, for self-reference; and type, to
 * allow renderers to determine correct printing aspects. These attributes
 * could be stored with the record in the database but this would result in
 * unnecessary storage as the values can be inferred from the load context.
 */

require_once('misc.php');

// different logical investment types
const INVESTMENT_TYPE_CANDIDATE = 0;       /* potential investment being contemplated
                                            * by the community */
const INVESTMENT_TYPE_PROPOSED = 1;        // currently being voted on
const INVESTMENT_TYPE_REJECTED = 2;        // proposed but didn't get majority vote
const INVESTMENT_TYPE_COMPLETE = 3;        // proposed and got majority vote

/**
 * Connect to the database.
 */
function db_init() {
    global $_CONFIG, $_TABLES, $GLOBALS;

    // include cassandra libraries
    $GLOBALS['THRIFT_ROOT'] = '/usr/share/php/Thrift';
    require_once $GLOBALS['THRIFT_ROOT'].'/packages/cassandra/Cassandra.php';
    require_once $GLOBALS['THRIFT_ROOT'].'/packages/cassandra/cassandra_types.php';
    require_once $GLOBALS['THRIFT_ROOT'].'/transport/TSocket.php';
    require_once $GLOBALS['THRIFT_ROOT'].'/protocol/TBinaryProtocol.php';
    require_once $GLOBALS['THRIFT_ROOT'].'/transport/TFramedTransport.php';
    require_once $GLOBALS['THRIFT_ROOT'].'/transport/TBufferedTransport.php';
    require_once('/usr/share/php/phpcassa/phpcassa.php');
    require_once('/usr/share/php/phpcassa/uuid.php');

    // register only standalone node
    CassandraConn::add_node($_CONFIG->cassandra_host, $_CONFIG->cassandra_port);

    // create globally accessible table instances
    define('TABLE_INVESTMENTS', 'Investments');
    define('TABLE_COMMENTS', 'Comments');
    define('TABLE_IMAGES', 'Images');
    define('TABLE_VOTES', 'Votes');
    define('TABLE_COMMENT_INDEX', 'CommentIndex');
    define('TABLE_BLOG', 'Blog');
    define('TABLE_LIKES', 'Likes');
    define('TABLE_SYSTEM', 'System');
    $ks = $_CONFIG->cassandra_keyspace;
    $_TABLES[TABLE_INVESTMENTS] = new CassandraCF($ks, TABLE_INVESTMENTS);
    $_TABLES[TABLE_COMMENTS] = new CassandraCF($ks, TABLE_COMMENTS);
    $_TABLES[TABLE_IMAGES] = new CassandraCF($ks, TABLE_IMAGES);
    $_TABLES[TABLE_VOTES] = new CassandraCF($ks, TABLE_VOTES);
    $_TABLES[TABLE_COMMENT_INDEX] = new CassandraCF($ks, TABLE_COMMENT_INDEX);
    $_TABLES[TABLE_BLOG] = new CassandraCF($ks, TABLE_BLOG);
    $_TABLES[TABLE_LIKES] = new CassandraCF($ks, TABLE_LIKES);
    $_TABLES[TABLE_SYSTEM] = new CassandraCF($ks, TABLE_SYSTEM);
}

/**
 * Store an investment in the database.
 *
 * First the image associated with the investment is stored in a separate
 * record, as a string of bytes. Secondly, because the database uses random
 * partitioning and therefore table scans are impossible, a record ID for the
 * investment is generated and stored in an 'index' record. Finally, the
 * investment data is stored as a record.
 *
 * @param string $title Investment title.
 * @param string $description Paragraph, or two, describing the investment.
 * @param string $charity_name Beneficiary charity's name.
 * @param string $charity_url URL to visit to learn more about the charity's
 * work.
 * @param string $author Username of investment author.
 * @param timestamp $created Investment created time.
 * @param string $img_data Bytes, as string, of PNG image associated with
 * investment.
 * @return string Investment record ID.
 */
function write_investment($title, $description, $charity_name, $charity_url,
                          $author, $created, $img_data) {
    global $_TABLES;
    
    // store image
    $record_id = uniqid();
    $_TABLES[TABLE_IMAGES]->insert($record_id, array('data' => $img_data));

    // append index
    $_TABLES[TABLE_INVESTMENTS]->insert(
        'index',
        array($record_id => $record_id));

    // store investment
    $_TABLES[TABLE_INVESTMENTS]->insert(
        $record_id,
        array(
            'id' => $record_id, 'title' => $title, 'description' => $description,
            'charity_name' => $charity_name, 'charity_url' => $charity_url,
            'author' => $author, 'created' => $created, 
            'comments' => '', 'type' => INVESTMENT_TYPE_CANDIDATE));
    return $record_id;
}

/**
 * Record a vote/like/dislike in the database.
 * @param string $table Database table, eg. TABLE_LIKES.
 * @param string $record_id Record ID of vote subject, ie. investment.
 * @param string $username Username of voter.
 * @param int $value Vote value, typically 1, but -1 for dislike. 
 */
function write_vote($table, $record_id, $username, $value=1) {
    global $_TABLES;
    $votes = $_TABLES[$table]->get($record_id);
    if (!empty($votes) && array_key_exists($username, $votes)) {
        error_log('User attempted to vote multiple times');
    } else {
       $_TABLES[$table]->insert($record_id, array($username => $value));
    }
}

/**
 * Write a comment to the database.
 *
 * Comments are stored in a separate database table to investments. A comments
 * index table is used to bind comment records to an investment. A record in
 * this index table with a key matching an investment record will hold an
 * array of associated comment records. Binding records in this way is thread-
 * safe.
 *
 * To maintain data integrity in a non-transaction database environment,
 * the comment record is written first, before the investment record
 * is updated. A database failure here could, as worst, produce an orphaned
 * comment record.
 *
 * @param array $investment Investment record being commented on.
 * @param string $author Username of comment author.
 * @param string $content Comment text.
 * @param timestamp $time Created UNIX timestamp.
 * @return array Comment record.
 */
function write_comment($investment, $author, $content, $time) {
    global $_TABLES;

    // create comment record
    $comment_id = uniqid();
    $record = array(
        'author' => $author,
        'content' => $content,
        'time' => $time);
    $_TABLES[TABLE_COMMENTS]->insert($comment_id, $record);

    // bind comment record to investment record
    // index records (columns) are sorted by key, so keys are timestamps
    // of when comment was made
    // 1 Jan 3000 = 32503680000 (11 chars)
    $padded_time = str_pad($time, 11, '0', STR_PAD_LEFT);
    $_TABLES[TABLE_COMMENT_INDEX]->insert($investment['id'],
                                          array($padded_time => $comment_id));
    return $record;
}

/**
 * Update existing investment database record.
 * @param array record
 */
function update_investment($record) {
    global $_TABLES;
    $_TABLES[TABLE_INVESTMENTS]->insert($record['id'], $record);
}

/**
 * Return the database record of a single investment (can be of any type).
 * @param string record ID
 * @return array record data
 */
function read_investment($record_id) {
    global $_TABLES;
    return $_TABLES[TABLE_INVESTMENTS]->get($record_id);
}

/**
 * Read investment image record.
 * @param string Record ID.
 * @return string Image bytes as string.
 */
function read_image($image_id) {
    global $_TABLES;
    return $_TABLES[TABLE_IMAGES]->get($image_id);
}

/**
 * Return a page of investments (winning candidates) ending on a set month.
 * @param timestamp inclusive month at end of range
 * @return array investment records
 */
function read_investments($to) {
    global $_TABLES, $_CONFIG;

    // get earliest requested month
    $from = subtract_month($to, $_CONFIG->investments_per_page);
    $start = strtotime($_CONFIG->investment_start);
    $month = max($from, $start);

    // get record key range for requested page of investments
    $key_range = array();
    while ($month <= $to) {
        $key_range[] = month_to_cassandra_key($month);
        $month = strtotime(date('Y-m', $month) . ' +1 month');
    }
    if (empty($key_range)) {
        return null;
    }

    // read all candidate record IDs within the range and reduce to array
    // containing only the IDs of winning candidates
    $candidates_range = $_TABLES[TABLE_BLOG]->multiget($key_range);
    foreach ($candidates_range as $candidates) {
        $winners[] = array_search(1, $candidates);
    }
    if (empty($winners)) {
        return array();
    }

    // adorn each investment with its database record (so that the data can
    // refer to itself (eg for like/dislike)
    $investments = $_TABLES[TABLE_INVESTMENTS]->multiget($winners);
    foreach ($investments as $key => &$investment) {
        $proposed = $investment['proposed'];
        $investment['page_id'] = strtotime(date('Y-m', intval($proposed))); // for pagination
    }

    // order investments by proposed date (reverse)
    usort($investments, function($a, $b) {
            return intval($b['proposed']) - intval($a['proposed']);
        });

    return $investments;
}

/**
 * Return the number of comment records associated with an investment record.
 * @param string $investment_id Investment record ID.
 * @return int Record count.
 */
function count_comments($investment_id) {
    global $_TABLES;
    return $_TABLES[TABLE_COMMENT_INDEX]->get_count($investment_id);
}

/**
 * Return like score for investment based on like/dislike votes. A like vote is
 * worth 1 point, a dislike vote is worth -1 point. The score is the sum of
 * all points.
 * @param string $investment_id Investment record ID.
 * @return int Score.
 */
function read_like_score($investment_id) {
    global $_TABLES;
    if (!$record = $_TABLES[TABLE_LIKES]->get($investment_id)) {
        return 0;
    }
    return array_sum(array_values($record));
}

/**
 * Read a page of comments belonging to an investment.
 * @param string contains ID of investment to which comments belong, and
 * zero-based index of first comment on page
 *
 * investment_id,start_index
 * eg 34a45def435aa,6
 * 
 * String needs to be contained in HTTP GET request for pagination.
 * String is assumed to have been validated at a higher level.
 */
function read_comments($page_id) {
    global $_TABLES, $_CONFIG;

    // unpack page ID
    $parts = explode(',', $page_id);
    $record_id = $parts[0];
    $last_comment_id = $parts[1];

    // get array of comment record IDs associated with investment record
    if (!$comment_index = $_TABLES[TABLE_COMMENT_INDEX]->get($record_id)) {
        return array(); // no comments
    }
    $comment_index = array_reverse(array_values($comment_index));

    // read a single page of comments (+1 to test for more comments)
    $page_start = intval(array_search($last_comment_id, $comment_index));
    $comment_keys = array_slice($comment_index, $page_start,
                                $_CONFIG->comments_per_page + 1);
    $comments = $_TABLES[TABLE_COMMENTS]->multiget($comment_keys);

    // extract and report missing comment records (which shouldn't be possible
    // given the order in which comment data is written
    $comments = array_filter($comments);
    if (count($comments) != count($comment_keys)) {
        error_log('Missing comment records detected');
    }

    // record IDs (keys) will be lost during sorting so add them to each
    // record as an attribute
    foreach ($comments as $key => &$comment) {
        $comment['record_id'] = $key;
    }
    
    // database returns requested comment records in random order so reorder
    // by created time (reverse chronological)
    usort($comments, function($a, $b) {
            return intval($b['time']) - intval($a['time']);
        });

    // mark each comment record with its page ID (required for pagination)
    foreach ($comments as &$comment) {
        $comment['page_id'] = $record_id . ',' . $comment['record_id'];
    }

    return $comments;
}

/**
 * Return all upcoming investment candidates stored in the database.
 * @return array upcoming investment data
 * @param int array index from which to read a page of investment records
 */
function read_upcoming_investments($from) {
    global $_TABLES, $_CONFIG;

    // read upcoming investment ID index
    if (!$upcoming_index = $_TABLES[TABLE_INVESTMENTS]->get('index')) {
        return null; // no investment candidates
    }

    // read a page of investments
    $upcoming_index = array_slice($upcoming_index, $from,
                                  $_CONFIG->candidates_per_page+1);

    // using index read individual upcoming invesments.
    $records = null;
    $i = 0;
    foreach ($upcoming_index as $upcoming_id) {
        $record = $_TABLES[TABLE_INVESTMENTS]->get($upcoming_id);
        // upcoming ID in index might not exist as row
        if ($record) {
            $record['page_id'] = $from + $i; // for pagination
            $records[] = $record;
        } else {
            error_log("Upcoming investment ID '$upcoming_id' not found");
        }
        $i++;
    }
    return $records;
}

/**
 * Return the 3 investment candidates associated with a month from the database.
 * @param timestamp month
 * @return array candidate records
 */
function read_candidates($month) {
    global $_TABLES;
    if (!$record_ids = read_candidate_ids($month)) {
        // can be null when app is first run without any candidate investments
        return null;
    }
    $candidates = $_TABLES[TABLE_INVESTMENTS]->multiget($record_ids);
    assert('!in_array(null, $candidates) /* candidates missing for month */');
    return $candidates;
}

/**
 * Read all votes made during a given month.
 * @param timestamp $month UNIX timestamp representation of month.
 * @return array Assoc array of votes ($record_id => array_of_usernames).
 */
function read_votes($month) {
    global $_TABLES;
    $record_ids = read_candidate_ids($month);
    return $_TABLES[TABLE_VOTES]->multiget($record_ids);
}

/**
 * Return the timestamp for the current month. It's useful to be able to control
 * this for simulations where time is accelerated.
 * @return timestamp Current month (as stored in database).
 */
function read_this_month() {
    global $_TABLES;
    $environment = $_TABLES[TABLE_SYSTEM]->get('environment');
    assert('array_key_exists(\'month\', $environment) /* month value not found in database */');
    $timestamp = strtotime($environment['month']);
    assert('$timestamp /* month value in database corrupt */');
    return $timestamp;
}

/**
 * Return the number of candidate investment records stored in the database.
 * @return int Record count.
 */
function read_num_candidates() {
    global $_TABLES;
    return $_TABLES[TABLE_INVESTMENTS]->get_count('index');
}

/**
 * Read ID of winning candidate investment for a given month.
 * @param timestamp $month
 * @return string Winning investment ID.
 */
function read_winning_candidate_id($month) {
    // assumes only one element will have a value of 1
    $winning_id = array_search(1, read_candidate_list($month));
    assert('$winning_id /* month doesn\'t have winning candidate */');
    return $winning_id;
}

/**
 * Read list of investment candidate IDs for a given month.
 * @param timestamp $month
 * @return array Array of investment record IDs.
 */
function read_candidate_ids($month) {
    $result = read_candidate_list($month);
    // can be empty when app is first run without any candidate investments
    assert('empty($result) || count($result) == 3 /* month has more than 3 candidates */');
    return empty($result) ? null : array_keys($result);
}

/**
 * Read list of investment candidate IDs for a given month. The list indicates
 * which, if any, candidate won (by setting its value to 1).
 * @param timestamp $month
 * @return array Assoc array of candidates (investment_id => 0|1).
 */
function read_candidate_list($month) {
    global $_TABLES;
    $key = month_to_cassandra_key($month);
    return $_TABLES[TABLE_BLOG]->get($key);
}

/**
 * Remove investment record from upcoming investment index, following it
 * being proposed to the community.
 * @param string $investment_id Investment record ID.
 */
function drop_investment_from_index($investment_id) {
    global $_TABLES;
    $_TABLES[TABLE_INVESTMENTS]->remove('index', $investment_id);
}

/**
 * Insert new month into blog or update existing month. Record is a list of
 * investment IDs and 0/1 to indicate whether the investment won the most
 * votes for the month.
 * @param string $key Month key, eg 199003.
 * @param array $record Assoc array (see above).
 */
function write_blog($key, $record) {
    global $_TABLES;
    $_TABLES[TABLE_BLOG]->insert($key, $record);
}

/**
 * Read environment variables.
 * @return array Record.
 */
function read_environment() {
    global $_TABLES;
    return $_TABLES[TABLE_SYSTEM]->get('environment');
}

/**
 * Write/update environment variables.
 * @param string $record Record.
 */
function write_environment($record) {
    global $_TABLES;
    $_TABLES[TABLE_SYSTEM]->insert('environment', $record);
}

?>
